package org.example.datastream.test;

import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.connector.jdbc.JdbcInputFormat;
import org.apache.flink.connector.jdbc.JdbcOutputFormat;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.StringUtils;

import java.sql.Types;

/**
 * Skeleton for a Flink Streaming Job.
 *
 * <p>For a tutorial how to write a Flink streaming application, check the
 * tutorials and examples on the <a href="https://flink.apache.org/docs/stable/">Flink Website</a>.
 *
 * <p>To package your application into a JAR file for execution, run
 * 'mvn clean package' on the command line.
 *
 * <p>If you change the name of the main class (with the public static void main(String[] args))
 * method, change the respective entry in the POM.xml file (simply search for 'mainClass').
 */
public class Job4 {
	public static void main(String[] args) throws Exception {
		final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

		new JdbcInputFormat();
		JdbcInputFormat input = JdbcInputFormat
				.buildJdbcInputFormat()
				.setDrivername("com.mysql.jdbc.Driver")
				.setDBUrl("jdbc:mysql://localhost:3306/flink-test?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
				.setUsername("root")
				.setPassword("123456")
				.setQuery("select * from t_users")
				.setRowTypeInfo(new RowTypeInfo(
						BasicTypeInfo.INT_TYPE_INFO,
						BasicTypeInfo.STRING_TYPE_INFO,
						BasicTypeInfo.STRING_TYPE_INFO))
				.finish();
		JdbcOutputFormat output = JdbcOutputFormat
				.buildJdbcOutputFormat()
				.setDrivername("com.mysql.jdbc.Driver")
				.setDBUrl("jdbc:mysql://localhost:3306/flink-test?useSSL=false&useUnicode=true&characterEncoding=UTF-8")
				.setUsername("root")
				.setPassword("123456")
				.setQuery("insert into t_users_copy(id, username, password) values(?, ?, ?)" +
						" on duplicate key update username=?, password=?")
				.setSqlTypes(new int[]{Types.INTEGER, Types.VARCHAR, Types.VARCHAR, Types.VARCHAR, Types.VARCHAR})
				.finish();

		env.createInput(input)
				.map(x -> {
					Object id = x.getField(0);
					Object username = x.getField(1);
					Object password = x.getField(2);

					password = StringUtils.byteToHexString(password.toString().getBytes());

					Row row = new Row(5);
					row.setField(0, id);
					row.setField(1, username);
					row.setField(2, password);
					row.setField(3, username);
					row.setField(4, password);
					return row;
				})
				.setParallelism(2)
				.writeUsingOutputFormat(output);

		// execute program
		env.execute("Flink Streaming Java API Skeleton");
	}
}
